package com.sec.etech.redis.stream.consumer;

import com.sec.etech.redis.stream.RedisStreamUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.openbpm.base.core.util.AppUtil;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Slf4j
//@Service
public class TestAsyncConsumeStreamListener2 implements StreamListener<String, ObjectRecord<String, Object>> {
    /**
     * 消费者类型：独立消费、消费组消费
     */
    private String consumerType;
    /**
     * 消费组
     */
    private String group;
    /**
     * 消费组中的某个消费者
     */
    private String consumerName;

    //private final RedisStreamUtil redisStreamUtil = AppUtil.getBean(RedisStreamUtil.class);
    @Resource
    private RedisStreamUtil redisStreamUtil;

    public TestAsyncConsumeStreamListener2() {

    }

    public TestAsyncConsumeStreamListener2(String consumerType, String group, String consumerName) {
        this.consumerType = consumerType;
        this.group = group;
        this.consumerName = consumerName;
    }

    @Override
    public void onMessage(ObjectRecord<String, Object> message) {
        log.info("######TestAsyncConsumeStreamListener2");
        String streamKey = message.getStream();
        RecordId recordId = message.getId();
        Object msg = message.getValue();

        if (StringUtils.isBlank(group)) {
            log.info("[{}]: 接收到一个消息 stream:[{}],id:[{}],value:[{}]", consumerType, streamKey, recordId, msg);
        } else {
            log.info("[{}] group:[{}] consumerName:[{}] 接收到一个消息 stream:[{}],id:[{}],value:[{}]", consumerType,
                    group, consumerName, streamKey, recordId, msg);
        }
        boolean b = this.consumeMessage(msg.toString());
        //是消费组消费并且消费成功
        // 当是消费组消费时，如果不是自动ack，则需要在这个地方手动ack
        if(b && StringUtils.isNotBlank(group)) {
            //redisStreamUtil.ack("streamKey",group,recordId.getValue());
        }
    }

    public boolean consumeMessage(String msg){
        log.info("consumeMessage消费消息，业务处理！");
        return true;
    }
}
